/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <climits>
#include <cstdio>
#include <cstdlib>
#include <fstream>
#include <iostream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include "BTreeIndex.h"

using namespace std;

// external functions and variables for load file and sql command parsing 
extern FILE* sqlin;
int sqlparse(void);


RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

bool SqlEngine::_check_cond(SelCond::Comparator comp, int diff)
{
  bool ret = true;

  switch (comp) {
    case SelCond::EQ:
      if (diff != 0)
        ret = false;
      break;
    case SelCond::NE:
      if (diff == 0)
        ret = false;
      break;
    case SelCond::GT:
      if (diff <= 0)
        ret = false;
      break;
    case SelCond::LT:
      if (diff >= 0)
        ret = false;
      break;
    case SelCond::GE:
      if (diff < 0)
        ret = false;
      break;
    case SelCond::LE:
      if (diff > 0)
        ret = false;
      break;
    default:
      ret = false;
      break;
  }

  return ret;
}

void SqlEngine::_print_tuple(int attr, int key, const string &value)
{
  switch (attr) {
    case 1:
      // SELECT key
      fprintf(stdout, "%d\n", key);
      break;
    case 2:
      // SELECT value
      fprintf(stdout, "%s\n", value.c_str());
      break;
    case 3:
      // SELECT *
      fprintf(stdout, "%d '%s'\n", key, value.c_str());
      break;
    default:
      // Do not print the tuple in other cases, e.g. SELECT COUNT(*).
      break;
  }
}

RC SqlEngine::_nonindex_scan(const string &table, const vector<SelCond> &cond, int attr)
{
  int count = 0;
  RC rc = 0;
  RecordFile rf;
  RecordId rid = {0, 0};

  // Open the table file.
  if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
    fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
    return rc;
  }

  // Scan the table file from the beginning.
  while (rid < rf.endRid()) {
    bool match = true;
    int key = 0;
    string value;

    // Read a tuple from the table file.
    if ((rc = rf.read(rid, key, value)) < 0) {
      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
      rf.close();
      return rc;
    }

    // Check the conditions on the tuple.
    for (unsigned i = 0; i < cond.size(); i++) {
      int diff = 0;

      // Compute the difference between the tuple and condition values.
      switch (cond[i].attr) {
        case 1:
          diff = key - atoi(cond[i].value);
          break;
        case 2:
          diff = strcmp(value.c_str(), cond[i].value);
          break;
      }

      // Skip the tuple if any condition is not met.
      if (!_check_cond(cond[i].comp, diff)) {
        match = false;
        break;
      }
    }

    // If all the conditions are met for the tuple, increment the matching
    // tuple counter and print the tuple if requested.
    if (match) {
      count++;
      _print_tuple(attr, key, value);
    }

    // Move to the next tuple.
    ++rid;
  }

  rf.close();

  return count;
}

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
  BTreeIndex index;
  IndexCursor cursor = {0, 0};
  RC rc = 0;
  RecordFile rf;
  bool useIndex = false;
  bool openedRf = false;
  bool nonKeyCond = false;    // Does there exist a condition on a value (i.e. non-key)?
  bool ineqCond = false;      // Does there exist an inequality condition on a key?
  bool eqCond = false;        // Does there exist an equality condition on a key?
  int eqSearchKey = 0;        // If there exists an equality condition on a key, we expect that key to be equal to this value.
  int minSearchKey = INT_MIN; // Key range lower limit.
  int maxSearchKey = INT_MAX; // Key range upper limit.
  int count = 0;

  // Scan the conditions to gather some stats (key range, etc.)
  for (unsigned i = 0; i < cond.size(); i++) {
    if (cond[i].attr == 1) {
      int searchKey = atoi(cond[i].value);
      if (cond[i].comp == SelCond::EQ) {
        if (eqCond && searchKey != eqSearchKey) {
          // Another equality condition already exists. These conditions
          // combined will always be false.
          goto exit_select;
        }
        eqCond = true;
        eqSearchKey = searchKey;
      }
      else if (cond[i].comp == SelCond::NE) {
        ineqCond = true;
      }
      else if (cond[i].comp == SelCond::GT) {
        if (searchKey + 1 > minSearchKey)
          minSearchKey = searchKey + 1;
      }
      else if (cond[i].comp == SelCond::LT) {
        if (searchKey - 1 < maxSearchKey)
          maxSearchKey = searchKey - 1;
      }
      else if (cond[i].comp == SelCond::GE) {
        if (searchKey > minSearchKey)
          minSearchKey = searchKey;
      }
      else if (cond[i].comp == SelCond::LE) {
        if (searchKey < maxSearchKey)
          maxSearchKey = searchKey;
      }
    }
    else if (cond[i].attr == 2) {
      nonKeyCond = true;
    }
  }

  // Ensure that the key range is valid.
  if (minSearchKey > maxSearchKey)
    goto exit_select;

  // If there exists an equality condition on the key, ensure that the equality
  // is within range. Then update the range to reflect the equality condition.
  if (eqCond) {
    if (eqSearchKey < minSearchKey || eqSearchKey > maxSearchKey)
      goto exit_select;
    minSearchKey = maxSearchKey = eqSearchKey;
  }

  // Decide whether to use simple table scan or index-assisted scan.
  // - If we want to scan the entire table:
  //   - If we care about values, then always use simple table scan.
  // - For all other cases, use index-assisted scan when possible.
  if (minSearchKey == INT_MIN && maxSearchKey == INT_MAX) {
    if (attr == 2 || attr == 3 || nonKeyCond) {
      rc = _nonindex_scan(table, cond, attr);
      if (rc < 0)
        goto exit_select_silent;
      else {
        count = rc;
        goto exit_select;
      }
    }
  }
  useIndex = (index.open(table + ".idx", 'r') == 0);
  if (!useIndex) {
    rc = _nonindex_scan(table, cond, attr);
    if (rc < 0)
      goto exit_select_silent;
    else {
      count = rc;
      goto exit_select;
    }
  }

  // Find the first B+Tree leaf node.
  if (index.locate(minSearchKey, cursor) != 0)
    goto exit_select;

  // Traverse the B+Tree leaf nodes.
  for (;;) {
next_key:

    RecordId rid = {0, 0};
    int key = 0;
    string value;

    if (index.readForward(cursor, key, rid) != 0)
      break;

    if (key < minSearchKey || key > maxSearchKey)
      break;

    // Check any inequality conditions on the key.
    if (ineqCond)
      for (unsigned i = 0; i < cond.size(); i++)
        if (cond[i].attr == 1 && cond[i].comp == SelCond::NE)
          if (key == atoi(cond[i].value))
            goto next_key;

    // Read the tuple ONLY WHEN NECESSARY!
    if (attr == 2 || attr == 3 || nonKeyCond) {
      // Open the table file if it is not yet open.
      if (!openedRf) {
        if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
          fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
          goto exit_select_silent;
        }
        openedRf = true;
      }

      // Read the tuple from the table file.
      if ((rc = rf.read(rid, key, value)) < 0) {
        fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
        goto exit_select_silent;
      }

      // Check any conditions on the value (non-key).
      if (nonKeyCond)
        for (unsigned i = 0; i < cond.size(); i++)
          if (cond[i].attr == 2)
            if (!_check_cond(cond[i].comp, strcmp(value.c_str(), cond[i].value)))
              goto next_key;

      // At this point, the key and value are stored in variables, ready to be
      // printed by _print_tuple().
    }

    // All the conditions are met for the tuple! Increment the matching
    // tuple counter and print the tuple if requested.
    count++;
    _print_tuple(attr, key, value);
  }

exit_select:

  // Print the matching tuple count if requested (i.e. if SELECT COUNT(*)).
  if (attr == 4)
    fprintf(stdout, "%d\n", count);

exit_select_silent:

  if (openedRf) {
    rf.close();
    openedRf = false;
  }

  if (useIndex) {
    index.close();
    useIndex = false;
  }

  return rc;
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
  if (table.empty()) {
    cerr << "Error: Please specify a table name." << endl;
    return RC_INVALID_ARGUMENTS;
  }

  if (loadfile.empty()) {
    cerr << "Error: Please specify a filename." << endl;
    return RC_INVALID_ARGUMENTS;
  }

  ifstream lf(loadfile.c_str(), ifstream::in);

  if (!lf) {
    cerr << "Error: Cannot open file '" << loadfile << "' for reading." << endl;
    return RC_FILE_OPEN_FAILED;
  }

  RecordFile rf(table + ".tbl", 'W');
  string line;

  BTreeIndex bti;

  if (index) {
    RC r;
    if ((r = bti.open(table + ".idx", 'W')) != 0) {
      rf.close();
      lf.close();
      cerr << "Error: Cannot open index file for writing." << endl;
      return r;
    }
  }

  while (getline(lf, line)) {
    int key;
    string value;

    if (parseLoadLine(line, key, value) != 0) {
      if (index)
        bti.close();
      rf.close();
      lf.close();
      cerr << "Error: Cannot parse file '" << loadfile << "'. Please check file format." << endl;
      return RC_FILE_READ_FAILED;
    }

    RecordId rid;

    rf.append(key, value, rid);

    if (index) {
      RC r;
      if ((r = bti.insert(key, rid)) != 0) {
        bti.close();
        rf.close();
        lf.close();
        cerr << "Error: Cannot insert into index." << endl;
        return r;
      }
    }
  }

  RC r1 = 0;
  if (index)
    r1 = bti.close();
  RC r2 = rf.close();
  lf.close();

  if (r1 == 0 && r2 == 0)
    return 0;
  else {
    cerr << "Warning: Error closing one or more files." << endl;
    return RC_FILE_CLOSE_FAILED;
  }
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;
    
    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');
    
    // if there is nothing left, set the value to empty string
    if (c == 0) { 
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}
